package rabbitmq

import (
	"encoding/json"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gorilla/websocket"
	"net"
	"time"
	"ws/server/config"
	"ws/server/model"
	"ws/server/websocket/clientManage"
)
import amqp "github.com/rabbitmq/amqp091-go"

type mq struct {
	conn       *amqp.Connection
	itpCh      *amqp.Channel
	wsCallBack *amqp.Channel

	itpSendQ    amqp.Queue
	itpReceiveQ amqp.Queue
	receiveMsg  <-chan amqp.Delivery

	notifyConnClose chan *amqp.Error
}

var Mq = mq{}

func Setup() {
	Mq.getConn()
	Mq.connect2MqChannel()
	Mq.notifyConnClose = Mq.conn.NotifyClose(make(chan *amqp.Error))
	g.Log().Info(config.Ctx, "mq 初始化完成")
}

func ReConn() {
	for Mq.notifyConnClose != nil {
		select {
		case err := <-Mq.notifyConnClose:
			g.Log().Warning(config.Ctx, "connection closed, error %s", err)
			Setup()
			time.Sleep(3 * time.Second)
		}
	}
}

func (m *mq) getConn() {
	conn, err := amqp.DialConfig(config.Rabbitmqaddr, amqp.Config{
		//Heartbeat: 30 * time.Second,
		Dial: func(network, addr string) (net.Conn, error) {
			return net.DialTimeout(network, addr, 60*time.Second)
		},
	})
	if err != nil {
		//g.Log().Error(ctx, err)
		time.Sleep(3 * time.Second)
		m.getConn()
	} else {
		m.conn = conn

		g.Log().Info(config.Ctx, "rabbitmq connection is established")
	}

}

func (m *mq) connect2MqChannel() {
	m.itpCh, _ = m.conn.Channel()
	m.itpReceiveQ, _ = m.itpCh.QueueDeclare(
		"",    // 队列名
		false, // 持久化队列，即使服务器重启，队列也会存在
		false, // 当队列不再被使用时，不要自动删除
		true,  // 不声明为独占队列，即其他连接也可以访问
		false, // 不使用no-wait模式，意味着声明操作会等待服务器确认
		nil,   // 额外的队列参数，这里没有使用
	)
	m.itpCh.QueueBind(
		m.itpReceiveQ.Name,     // queue name
		"",                     // routing key
		config.ItpExchangeName, // exchange
		false,
		nil,
	)
	m.receiveMsg, _ = m.itpCh.Consume(
		Mq.itpReceiveQ.Name, // 使用上面声明的队列名称
		"",                  // 不指定消费者标签，RabbitMQ会自动生成一个
		true,                // 自动确认消息
		false,               // 不声明为独占消费者
		false,               // 允许将消息传递给本机上的其他消费者
		false,               // 不使用no-wait模式
		nil,                 // 消费者参数，这里没有使用
	)
}

func ReceiveMsg() {
	for {
		for d := range Mq.receiveMsg {
			callbacks := model.MsgObject{}
			g.Dump(d.Body)
			err := json.Unmarshal(d.Body, &callbacks)
			if err != nil {
				g.Log().Error(config.Ctx, "解码消息出错:", err)
				continue
			}
			//g.Log().Info(config.Ctx, "收到的消息体：", callbacks)
			// 根据条件筛选消息
			if callbacks.Type == "txtMessage" {
				// 广播消息到特定的前端
				err := clientManage.BroadcastToChannel(callbacks.Topic, websocket.TextMessage, callbacks, callbacks.TargetCodes)
				if err != nil {
					g.Log().Error(config.Ctx, "Broadcast error:", err)
				}
			}
		}
		g.Log().Error(config.Ctx, "receive queue  is closed . waiting to reconnect")
		time.Sleep(time.Second)
	}
}
